package com.xxmw.transaction.controller.server;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Sequence;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.tomcat.websocket.WsSession;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author luowenliang
 * @since 2020/3/11 14:20
 */
@Slf4j
@Component
@ServerEndpoint("/ws/transaction")
public class TransactionWebsocketServer {

    Sequence sequence = new Sequence();

    /**
     * 统计当前连接的客户端数量
     */
    private static AtomicInteger onlineNum;

    /**
     * 存放客户端
     */
    private static ConcurrentHashMap<Long, TransactionWebsocketServer> clientContainer;

    private static final String TOPIC_SUB_FIELD = "sub";

    /**
     * 订阅主题
     */
    private Set<String> subscribedTopic;

    private Session currentSession;

    private Long currentClientId;

    static {
        onlineNum = new AtomicInteger(0);
        clientContainer = new ConcurrentHashMap<>();
    }

    {
        subscribedTopic = new HashSet<>();
    }

    /**
     * 连接建立成功调用的方法
     * @param session
     * @param clientId @PathParam("clientId") String clientId
     */
    @OnOpen
    public void onOpen(Session session) {
        WsSession wsSession = (WsSession) session;
        log.error("session id: {}, HttpSession id:{}", wsSession.getId(), wsSession.getHttpSessionId());
        this.currentSession = session;
        this.currentClientId = sequence.nextId();
        clientContainer.putIfAbsent(this.currentClientId, this);
        this.onlineAutoIncrement();
        log.info("当前连接客户端: {} , 当前连接数: {} ", this.currentClientId, onlineNum.get());
//            sendMessage("【WS服务端通知】：已连接成功");
//        sendMessage(JSON.toJSONString(SpringContextUtil.getBean(RobotOrderService.class).volumeStatistics()));
    }

    /**
     * 客户端关闭时调用的方法
     */
    @OnClose
    public void OnClose () {
        if(clientContainer.containsKey(currentClientId)){
            clientContainer.remove(currentClientId);
            this.onlineAutoDecrement();
        }
        log.info("客户端: {} 退出,当前连接数: {} ", currentClientId, onlineNum.get());
    }

    /**
     * 收到客户端信息的时候调用的方法
     *
     *  订阅主题 { "sub": "topic to sub"}
     *  取消订阅 { "unsub": "topic to unsub"}
     */
    @OnMessage
    public void OnMessage (Session session, String message) {
        log.info("收到客户端： {} 的消息： {} ", currentClientId, message);
        if (StringUtils.isNotBlank(message)) {
            if (message.contains(TOPIC_SUB_FIELD)) {
                //订阅消息
                subscribed(message);
            } else {
                log.error("未处理收到消息");
            }
        }
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("服务端错误", error);
    }

    /**
     * 实现服务器主动推送
     * @param message
     * @throws IOException
     */
    private void sendMessage(String message) {
        try {
            this.currentSession.getBasicRemote().sendText(message);
        } catch (IOException e) {
            log.error("发送消息异常", e);
        }
    }

    /**
     * 订阅失败发送消息
     * @param message
     */
    public void sendErrorMessage (String message) {
        JSONObject res = new JSONObject();
        res.put("error", message);
        res.put("state", 0);
        sendMessage(res.toJSONString());
    }

    /**
     * 订阅成功发送消息
     * @param message
     */
    public void sendSuccessMessage (String message) {
        JSONObject res = new JSONObject();
        res.put("topic", message);
        res.put("state", 1);
        sendMessage(res.toJSONString());
    }

    /**
     * 发送数据方法
     * @param key
     * @param data
     */
    public void senObject (Object data) {
        JSONObject object = new JSONObject();
        object.put("result", data);
        this.sendMessage(object.toJSONString());
    }

    /**
     * 订阅主题
     * @param message
     */
    private void subscribed (String message) {
        JSONObject jsonObject = JSON.parseObject(message);
        String topic = jsonObject.getString(TOPIC_SUB_FIELD);
        if (null != topic) {
            subscribedTopic.add(topic);
            sendSuccessMessage(topic);
            log.error("客户端：{} 订阅主题: {} 成功", this.currentClientId, topic);
        } else {
            sendErrorMessage("topic can not be empty");
        }
    }

    private int onlineAutoIncrement () {
        return onlineNum.incrementAndGet();
    }

    private int onlineAutoDecrement () {
        return onlineNum.decrementAndGet();
    }

    public static ConcurrentHashMap<Long, TransactionWebsocketServer> getServer () {
        return clientContainer;
    }

    public boolean isSubscribed (String topic) {
        return subscribedTopic.contains(topic);
    }
}
